package com.desmart.cdc.mysql;


import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.desmart.cdc.common.MyDeserialization;
import com.ververica.cdc.connectors.mysql.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.util.Collector;

/**
 * 使用自定义序列化器和自定义处理过程
 */
public class MysqlCDC2 {


    public static void main(String[] args) throws Exception {
        SourceFunction<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("47.103.71.206")
                .port(3306)
                .databaseList("cdc_source") // set captured database, If you need to synchronize the whole database, Please set tableList to ".*".
                .tableList("cdc_source.user") // set captured table
                .username("root")
                .password("Passw0rd!")
                .deserializer(new MyDeserialization()) // converts SourceRecord to JSON String
                .startupOptions(StartupOptions.initial())  // 从哪里开始
                .build();

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // enable checkpoint
        //env.enableCheckpointing(3000);

        DataStreamSource<String> stringDataStreamSource = env
                .addSource(mySqlSource)
                // set 4 parallel source tasks
                .setParallelism(1);

        stringDataStreamSource.process(new ProcessFunction<String, Object>() {
            @Override
            public void processElement(String s, Context context, Collector<Object> collector) throws Exception {
                //System.out.println("process: " + s);
                JSONObject jsonObject = JSON.parseObject(s);
                String op = jsonObject.getString("op");
                Envelope.Operation operation = Envelope.Operation.forCode(op);
                if (operation == Envelope.Operation.READ) {
                    System.out.println("初始化读取数据：" + jsonObject.getJSONObject("after").toJSONString());
                } else if (operation == Envelope.Operation.CREATE) {
                    System.out.println("新增数据：" + jsonObject.getJSONObject("after").toJSONString());
                } else if (operation == Envelope.Operation.UPDATE) {
                    System.out.println("更新数据：" + jsonObject.getJSONObject("after").toJSONString());
                } else if (operation == Envelope.Operation.DELETE) {
                    System.out.println("删除数据：" + jsonObject.getJSONObject("before").toJSONString());
                 }else if (operation == Envelope.Operation.TRUNCATE) {
                    System.out.println("清空数据：");
                }

            }
        }).setParallelism(1);
        env.execute("Print MySQL Snapshot + Binlog");
    }

}
